package seven.six;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import seven.entity.InfoBean;

public class JoinMR {
	static class JoinMRMapper extends Mapper<LongWritable,Text,Text,InfoBean>{
		InfoBean bean = new InfoBean();
		Text k = new Text();
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, InfoBean>.Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] fields = line.split("\t");
			FileSplit inputSplit = (FileSplit)context.getInputSplit();
			String filename = inputSplit.getPath().getName();
			String pId = "";
			if(filename.startsWith("order")){
				pId = fields[2];
				bean.set(Integer.parseInt(fields[0]), fields[1], pId, Integer.parseInt(fields[3]), "", 0, 0,"0");
			}else{
				pId = fields[0];
				bean.set(0,"", pId, 0,fields[1], Integer.parseInt(fields[2]),Integer.parseInt(fields[3]),"1");
			}
			k.set(pId);
			context.write(k,bean);
		} 
	}
	static class JoinMRReducer extends  Reducer<Text,InfoBean,InfoBean,NullWritable>{

		@Override
		protected void reduce(Text pId, Iterable<InfoBean> beans,
				Reducer<Text, InfoBean, InfoBean, NullWritable>.Context context) throws IOException, InterruptedException {
			 InfoBean pdBean = new InfoBean();
			 ArrayList<InfoBean> orderBeans =  new ArrayList<InfoBean>();
			 try {
				for(InfoBean bean : beans){
					if("1".equals(bean.getFlag())){
						BeanUtils.copyProperties(pdBean, bean);
					}else{
						InfoBean odbean = new InfoBean();
						BeanUtils.copyProperties(odbean, bean);
						orderBeans.add(odbean);
					}
				}
			} catch (Exception e) {
				// TODO: handle exception
			}
			 
			 
			for(InfoBean bean : orderBeans){
				bean.setPname(pdBean.getPname());
				bean.setCategoryId(pdBean.getCategoryId());
				bean.setPrice(pdBean.getPrice());
				context.write(bean,NullWritable.get());
			}
		} 
	}
	public static void main(String[] args)throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(JoinMR.class);
		
		job.setMapperClass(JoinMRMapper.class);
		job.setReducerClass(JoinMRReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(InfoBean.class);
		
		job.setOutputKeyClass(InfoBean.class);
		job.setOutputValueClass(NullWritable.class);
		
		FileInputFormat.setInputPaths(job,new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/input/"));
		//指定job输出结果目录所在
		FileOutputFormat.setOutputPath(job, new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/output/"));
		
		boolean res = job.waitForCompletion(true);
		System.exit(res?0:1);
		
		
	}
}
